package com.syuchen.client.udp;

import com.syuchen.common.Invocation;
import com.syuchen.common.URL;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.Map;
import java.util.concurrent.*;

/**
 * udp socket池
 * 自动通过心跳维护udp路由，空闲超过一定时间则关闭
 * 循环检测每个socket，将收到的消息放到生产者-消费者队列中
 */
public class UdpSocketPool {

    private static ProduceConsumeQueue<String> msgQueue = new ProduceConsumeQueue<>(Integer.MAX_VALUE);

    private static final Map<URL, DatagramSocket> socketMap = new ConcurrentHashMap<>();

    private static final long HEARTBEAT_TIME = 30; // 心跳周期：30秒

    private static ScheduledExecutorService heartbeatExecutor  = Executors.newScheduledThreadPool(1);

    //使用多线程io 在多核的情况下更快
    private static ExecutorService receiveMsgExecutor = Executors.newFixedThreadPool(10);

    public static synchronized DatagramSocket getDatagramSocket(URL url) throws IOException {
        DatagramSocket datagramSocket = socketMap.get(url);
        if (socketMap.get(url) != null) {
            System.out.println("当前端口：" + datagramSocket.getPort());
            return datagramSocket;
        } else {
            DatagramSocket newSocket = new DatagramSocket();
            socketMap.put(url, newSocket);
            startSocketMsgReceive(url);
            startSocketHeartbeat(url);
            return newSocket;
        }
    }

    private static void startSocketMsgReceive(final URL url) {
        final DatagramSocket datagramSocket = socketMap.get(url);
        receiveMsgExecutor.execute(() -> {
            try {
                while (true) {
                    DatagramPacket datagramPacket = new DatagramPacket(new byte[1200], 1200);
                    datagramSocket.receive(datagramPacket);// receive方法内部加了锁
                    byte[] arr = datagramPacket.getData();
                    int len = datagramPacket.getLength();
                    msgQueue.put(new String(arr, 0, len));
                }
            } catch (IOException ignored) {
            } finally {
                socketMap.remove(url);
                datagramSocket.close();
            }
        });
    }

    /**
     * 创建一个新的DatagramSocket时启动一个针对它的周期心跳任务，以维护udp路由
     */
    private static void startSocketHeartbeat(final URL url) {
        final DatagramSocket datagramSocket = socketMap.get(url);
        heartbeatExecutor.scheduleAtFixedRate(() -> {
            try {
                String msg = Invocation.HEARTBEAT_MSG;
                datagramSocket.send(new DatagramPacket(msg.getBytes() ,msg.length(), InetAddress.getByName(url.getHostname()), url.getPort()));
            } catch (Exception e) {
                socketMap.remove(url);
                datagramSocket.close();
                e.printStackTrace();
            }
        }, HEARTBEAT_TIME, HEARTBEAT_TIME, TimeUnit.SECONDS);
    }

    public static String takeMsg() throws InterruptedException {
        return msgQueue.take();
    }

    public static void stopMsg() {
        receiveMsgExecutor.shutdownNow();
        receiveMsgExecutor = Executors.newFixedThreadPool(10);
        heartbeatExecutor.shutdownNow();
        heartbeatExecutor  = Executors.newScheduledThreadPool(1);
        socketMap.forEach((u, s) -> s.close());
        socketMap.clear();
    }
}
